package com.wutoon.etl.handler;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.wutoon.etl.aop.PowerJobLogReport;
import com.wutoon.etl.constant.CommonConstant;
import com.wutoon.etl.handler.function.DealResponseFunction;
import com.wutoon.etl.mapper.BizOperateRecordManage;
import com.wutoon.etl.pojo.AviatorResult;
import com.wutoon.etl.pojo.Db2HttpTask;
import com.wutoon.etl.pojo.Http2DbTask;
import com.wutoon.etl.util.AviatorUtils;
import com.wutoon.etl.util.CommonUtil;
import com.wutoon.etl.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;

import java.lang.reflect.InvocationTargetException;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicMarkableReference;

/**
 * @author 武兴云/72176468
 * @version 1.0
 * @date 2024/1/22 17:48
 * @description http方式推送数据
 */
@Slf4j
public class HttpPusher implements Runnable{
    private final BlockingQueue<Object[]> queue;

    private final Db2HttpTask task;

    private final CountDownLatch consumerEnd;

    private final AtomicMarkableReference<String> msg;

    private final RestTemplate restTemplate;

    private final CountDownLatch producerEnd;

    private final BizOperateRecordManage bizOperateRecordManage;

    private final ConcurrentHashMap<Integer, String> columnMap;

    private RequestAttributes requestAttributes;

    public HttpPusher(RestTemplate restTemplate, RequestAttributes requestAttributes, CountDownLatch producerEnd,
                      BlockingQueue<Object[]> queue, ConcurrentHashMap<Integer, String> columnMap, Db2HttpTask task,
                      CountDownLatch consumerEnd, AtomicMarkableReference<String> msg, BizOperateRecordManage bizOperateRecordManage) {
        this.queue = queue;
        this.columnMap = columnMap;
        this.task = task;
        this.consumerEnd = consumerEnd;
        this.msg = msg;
        this.restTemplate = restTemplate;
        this.producerEnd = producerEnd;
        this.bizOperateRecordManage = bizOperateRecordManage;
        this.requestAttributes = requestAttributes;
    }

/**
 *
 * @author 武兴云/72176468
 * @date 2024/2/29 9:05
 * @update_by 武兴云/72176468
 * @update_at 2024/2/29 9:05
 * @creed if you have doubt , please contact me !!!
 */
@Override
public void run() {
    log.info("http pusher start ......");
    long dataSize = 0;
    long start = System.currentTimeMillis();
    // 子线程放入requestAttributes，方便上报日志
    RequestContextHolder.setRequestAttributes(requestAttributes);
    PowerJobLogReport.report("http pusher start ......");
    try {
        // 在此处并发调用http接口
        Db2HttpTask.HttpPushSourceConfig sourceConfig = task.getTargetSourceConfig();
        // 有数据，并且主线程没有停止子线程
        List<JSONObject> data = new ArrayList<>();
        while (true) {
            // 主线程终止了子线程
            if (Thread.currentThread().isInterrupted()) {
                break;
            }
            Object[] poll = queue.poll(CommonConstant.POLL_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            // 生产者执行结束 并且 队列为空
            if (poll == null && producerEnd.getCount() < CommonConstant.NUMBER_ONE) {
                break;
            }
            // 获取为空继续下一轮判断
            if (poll == null) {
                continue;
            }
            // 不同数据库之间etl，需要注意字段类型转换，eg：oracle的timestamp转ob的timestamp，查询时使用cast(xxx as date)
            JSONObject item = new JSONObject();
            for (int i = 0; i < poll.length; i++) {
                item.put(columnMap.get(i), poll[i]);
            }
            data.add(item);
            dataSize ++;
            if (dataSize % task.getBatchInsertSize() == 0) {
                log.info("HttpPusher execute batch push:{}, data size:{}", task.getBatchInsertSize(), dataSize);
                PowerJobLogReport.report("HttpPusher execute batch push:%s, data size:%s", task.getBatchInsertSize(), dataSize);
                pushData(data, sourceConfig);
                data.clear();
            }
        }
        if (CollectionUtils.isNotEmpty(data)) {
            pushData(data, sourceConfig);
            data.clear();
        }
    } catch (Exception e) {
        log.error("http pusher error:, ", e);
        String stackTraceString = CommonUtil.getStackTraceString(e);
        msg.set(stackTraceString, true);
    } finally {
        consumerEnd.countDown();
        log.info("http pusher end duration:{}, dataSize:{}", System.currentTimeMillis() - start, dataSize);
        PowerJobLogReport.report("http pusher end duration:%s, dataSize:%s", System.currentTimeMillis() - start, dataSize);
        // resetRequestAttributes
        RequestContextHolder.resetRequestAttributes();
    }
}

    private void pushData(List<JSONObject> data, Db2HttpTask.HttpPushSourceConfig sourceConfig) {
        // 组装请求头
        HttpHeaders headers = buildHeaders(sourceConfig);
        // 组装请求体
        JSONObject bodyTemplate = sourceConfig.getBodyTemplate() == null ? new JSONObject() : sourceConfig.getBodyTemplate();
        String bizOperateRecordId = IdUtil.getSnowflakeNextIdStr();

        String realParam;
        if (StringUtils.equals(sourceConfig.getData().getPosition(), CommonConstant.POSITION_ROOT)) {
            realParam = JSON.toJSONString(data, SerializerFeature.WriteMapNullValue);
        } else {
            bodyTemplate.put(sourceConfig.getData().getKey(), data);
            if (sourceConfig.getRequestNo() != null && StringUtils.isNotEmpty(sourceConfig.getRequestNo().getKey())) {
                // 请求批次号
                bodyTemplate.put(sourceConfig.getRequestNo().getKey(), bizOperateRecordId);
            }
            realParam = JSON.toJSONString(bodyTemplate, SerializerFeature.WriteMapNullValue);
        }

        // 请求远程接口
        long beforeSend = System.currentTimeMillis();
        ResponseEntity<JSONObject> responseEntity = null;
        try {
            // 替换参数模板
            HttpEntity<String> requestEntity;
            if (Objects.equals(HttpMethod.resolve(sourceConfig.getMethod()), HttpMethod.GET)) {
                requestEntity = new HttpEntity<>(null, headers);
            } else {
                requestEntity = new HttpEntity<>(realParam, headers);
            }
            // 发送请求
            responseEntity = restTemplate.exchange(sourceConfig.getUrl(), Objects.requireNonNull(HttpMethod.resolve(sourceConfig.getMethod())),
                    requestEntity, JSONObject.class);
            // 判断是否成功
            JSONObject responseBody = responseEntity.getBody();
            AviatorResult<Boolean> successResult = AviatorUtils.executeExpression(sourceConfig.getJudgeSuccessExpression(), responseBody, Boolean.class);
            if (successResult.getResult() == null || successResult.getResult().equals(Boolean.FALSE)) {
                throw new RuntimeException("push data error");
            }
            // 观察该日志，看数据队列的实时容量，判断是生产者和消费者是否平衡。如果容量偏满则增加消费者
            log.info("Thread-name:{},current push data,queue size:{}",Thread.currentThread().getName(), queue.size());
        } catch (Exception e) {
            log.error("http collector request remote error,param:{}, exception:", sourceConfig, e);
            String stackTraceString = CommonUtil.getStackTraceString(e);
            msg.set(stackTraceString, true);
        } finally {
            // 请求记录保存
            long afterSend = System.currentTimeMillis();
            bizOperateRecordManage.saveBizOperateRecord(responseEntity, realParam, bizOperateRecordId, DateUtil.generateCurrentDayId(DateUtil.YYYYMMDD), beforeSend, afterSend, sourceConfig.getSystem());
        }

        // 发生异常退出
        if (!msg.isMarked()) {
            log.info("HttpPusher push batch success:{}", data.size());
        } else {
            log.error("HttpPusher end error:{}", msg.getReference());
            throw new RuntimeException(msg.getReference());
        }
    }

    /**
     * 本剧入参组装请求头
     *
     * @param sourceConfig Http2DbTHttpCollectSourceConfig
     * @return HttpHeaders
     * @author 武兴云/72176468
     * @date 2024/2/28 15:13
     * @update_by 武兴云/72176468
     * @update_at 2024/2/28 15:13
     * @creed if you have doubt , please contact me !!!
     */
    private static HttpHeaders buildHeaders(Db2HttpTask.HttpPushSourceConfig sourceConfig) {
        HttpHeaders headers = new HttpHeaders();
        JSONObject head = sourceConfig.getHead();
        head.forEach((key, value) -> headers.add(key, (String) value));
        if (!headers.containsKey(HttpHeaders.CONTENT_TYPE)) {
            headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
        }
        return headers;
    }


    private List<Object> unStructResponse(Http2DbTask.HttpCollectSourceConfig sourceConfig, JSONObject responseBody) {
        if (sourceConfig.getDataExpression().startsWith(CommonConstant.REF_METHOD)) {
            String functionName = sourceConfig.getDataExpression().split(CommonConstant.REF_METHOD)[1];
            try {
                Class<?> function = Class.forName(functionName);
                DealResponseFunction instance = (DealResponseFunction) function.getDeclaredConstructor().newInstance();
                return instance.unStruct(responseBody);
            } catch (ClassNotFoundException | NoSuchMethodException | InstantiationException | IllegalAccessException |
                     InvocationTargetException e) {
                throw new RuntimeException(e);
            }
        } else {
            AviatorResult<ArrayList> dataResult = AviatorUtils.executeExpression(sourceConfig.getDataExpression(), responseBody, ArrayList.class);
            return dataResult.getResult();
        }
    }



}
